package com.zang.websocket.server;

import com.zang.netty.model.enums.SocketEnum;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.handler.stream.ChunkedWriteHandler;
import io.netty.util.CharsetUtil;

import java.nio.charset.Charset;
import java.util.Date;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

import static io.netty.handler.codec.http.HttpHeaderUtil.isKeepAlive;
import static io.netty.handler.codec.http.HttpHeaderUtil.setContentLength;
import static io.netty.handler.codec.http.HttpResponseStatus.BAD_REQUEST;
import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1;

/**
 * netty webSocket server
 *
 * @Description TODO
 * @Author ZhangQiang
 * @Date 2019-7-21 10:13
 **/
public class WebSocketServer {

    public static void main(String[] args) throws InterruptedException {
        int port = SocketEnum.PORT.getNum();
        new WebSocketServer().run(port);
    }


    public void run(int port) throws InterruptedException {

        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();

        try {

            ServerBootstrap serverBootstrap = new ServerBootstrap();

            serverBootstrap.group(bossGroup, workGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChildChannelInitializer());


            Channel channel = serverBootstrap.bind(port).sync().channel();

            System.out.println(
                    "websocket server started at port " + port + "."
            );
            System.out.println(
                    "Open your browser and navigate to http://localhost:" + port + '/'
            );

            channel.closeFuture().sync();

        } finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }
}

/**
 * 初始化配置
 **/
class ChildChannelInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast("http-codec", new HttpServerCodec());
        pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
        socketChannel.pipeline().addLast("http-chunked", new ChunkedWriteHandler());
        pipeline.addLast("handler", new WebSocketServerHandler());
    }
}

/**
 *
 **/


/**
 * webSocketServer 处理类
 **/
class WebSocketServerHandler extends SimpleChannelInboundHandler<Object> {

    private WebSocketServerHandshaker handshaker;

    /**
     *
     * 创建Socket连接需要客户端先向服务端发起一个Http请求，这个请求包含附加头信息：Upgrade:WebSocket，表明这是一个升级的Http协议请求，也就是WebSocket
     * 服务器解析后生成应答信息返回给客户端，这样双向的连接就建立了，此连接会持续到双方其一主动关闭连接
     * 其中请求头中会包含一个Sec-WebSocket-Key
     * 返回头中会包含一个Sec-WebSocket-Accept
     * Sec-WebSocket-Accept 是Sec-WebSocket-Key加上一个魔幻字符串然后使用SHA-1加密然后进行BASE-64编码，然后结果作为Sec-WebSocket-Accept的值返回给客户端
     *
     */
    @Override
    protected void messageReceived(ChannelHandlerContext channelHandlerContext, Object o) throws Exception {
        System.out.println("解析请求");
        // http 请求 第一次握手请求消息由HTTP协议承载，所以它是一个HTTP消息，执行handleHttpRequest方法来处理WebSocket握手请求
        if (o instanceof FullHttpRequest) {
            handleHttpRequest(channelHandlerContext, (FullHttpRequest) o);
        }
        //WebSocket请求  客户端通过文本框提交请求消息给服务端，WebSocketServerHandler接收到的是已经解码后的WebSocketFrame消息
        if (o instanceof WebSocketFrame) {
            handleWebSocketFrame(channelHandlerContext, (WebSocketFrame) o);
        }

    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    // 解码http请求
    private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {

        // 如果http解码失败 返回http异常
        if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {
            sentHttpResponse(ctx, req, new DefaultFullHttpResponse(HTTP_1_1, BAD_REQUEST));
            System.out.println("解析异常");
            return;
        }
        String url = "ws://localhost:" + SocketEnum.PORT.getNum() + "/websocket";
        System.out.println("url:" + url);
        // 握手工厂 构造握手响应返回 本地测试
        WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(url, null, false);
        handshaker = wsFactory.newHandshaker(req);
        if (null == handshaker) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        } else {
            handshaker.handshake(ctx.channel(), req);
        }

    }

    // 解码websocket请求
    private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) throws ExecutionException, InterruptedException {
        // 判断是否是关闭链路的指令
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }
        // 判断是否是ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        //判断如果不是文本消息则抛出异常
        if (!(frame instanceof TextWebSocketFrame)) {
            throw new UnsupportedOperationException(String.format("%s frame types noe supported", frame.getClass().getName()));
        }

        String request = ((TextWebSocketFrame) frame).text();
        System.out.println("request :" + request);
        if (request.contains("+")) {
            String a = request.substring(0, request.indexOf("+"));
            String b = request.substring(request.indexOf("+") + 1, request.length());
            Integer ia = Integer.parseInt(a);
            Integer ib = Integer.parseInt(b);

            //CompletableFuture 异步执行获取结果
            //thenApplyAsync 返回结果
            String c = CompletableFuture.supplyAsync(() -> {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return ia + ib;
            }).thenApplyAsync(v -> v + " = " + ia + " + " + ib).join();

            //消费结果
            CompletableFuture.supplyAsync(() -> {
                return ia + ib;
            }).thenAccept(v -> {
                System.out.println(v);
            });

            //get()
            Integer d = CompletableFuture.supplyAsync(() -> {
                return ia + ib;
            }).get();


            ctx.channel().write(
                    new TextWebSocketFrame("计算结果 ： " + c));
        } else {
            ctx.channel().write(
                    new TextWebSocketFrame("输入内容为：" + request + ", <br/>欢迎使用Netty WebSocket 服务，现在时刻：" + new Date().toString()));
        }
    }


    //发送响应
    private static void sentHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
        // 返回响应给客户端
        if (res.status().code() != 200) {
            ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
            res.content().writeBytes(buf);
            buf.release();
            setContentLength(res, res.content().readableBytes());
        }

        // 如果是非Keep-Aliv，关闭连接
        ChannelFuture f = ctx.channel().writeAndFlush(res);
        if (!isKeepAlive(req) || res.status().code() != 200) {
            f.addListener(ChannelFutureListener.CLOSE);
        }
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}